package com.boilermaker.mq.mqserver;

import com.boilermaker.mq.common.Consumer;
import com.boilermaker.mq.common.ConsumerEnv;
import com.boilermaker.mq.common.MqException;
import com.boilermaker.mq.mqserver.core.*;
import com.boilermaker.mq.mqserver.manager.ConsumeManager;
import com.boilermaker.mq.mqserver.manager.DiskDataCenter;
import com.boilermaker.mq.mqserver.manager.MemoryDataCenter;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/*
 * VirtualHost 是逻辑上的数据集合，使用它区分不同业务线的数据，
 * 每个虚拟主机都独立管理自己的交换机、队列、绑定和消息数据，保证隔离性。
 * 该类将进一步封装‘内存数据操作’和‘硬盘数据操作’，提供该 MQ 程序的核心 API，并统一处理下层代码抛出的异常。
 */
@Slf4j
@Getter
public class VirtualHost {
    private final String virtualHostName;
    private final MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private final DiskDataCenter diskDataCenter = new DiskDataCenter();
    private final Router router = new Router();
    // 与 ConsumeManager 相互关联
    private final ConsumeManager consumeManager = new ConsumeManager(this);
    // 锁对象
    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();

    // 构造方法：用于初始化或重启程序
    public VirtualHost(String name) {
        this.virtualHostName = name;
        // 若是初次启动程序，则建库建表，若此时已有数据库，init 则不会做任何事
        diskDataCenter.init();
        // 保证内存数据和硬盘数据的统一
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            log.info("恢复内存数据失败, virtualHostName=" + name);
            e.printStackTrace();
        }
    }

    // 创建交换机
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType,
                                   boolean durable, boolean autoDelete, Map<String, Object> arguments) {
        // 为 exchangeName 加上所在虚拟主机的前缀，以保证不同虚拟主机中 exchangeName 的命名空间一致
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 查询操作尽量查内存数据，效率更高。（硬盘数据仅用于持久化存储）
                Exchange toDeclare = memoryDataCenter.getExchange(exchangeName);
                // 若存在则直接返回
                if (toDeclare != null) {
                    log.info("交换机已经存在, exchangeName=" + exchangeName);
                    return true;
                }
                // 若不存在则创建交换机
                Exchange newExchange = new Exchange();
                newExchange.setName(exchangeName);
                newExchange.setType(exchangeType);
                newExchange.setDurable(durable);
                newExchange.setAutoDelete(autoDelete);
                newExchange.setArguments(arguments);
                // 硬盘存储（在该交换机需要持久化存储的时候）
                // 先存硬盘，因为硬盘存储失败的情况要多于内存存储失败的情况
                if (durable) diskDataCenter.insertExchange(newExchange);
                // 内存存储
                memoryDataCenter.insertExchange(newExchange);
                log.info("交换机创建完成, exchangeName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            log.info("交换机创建失败, exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除交换机
    public boolean exchangeDelete(String exchangeName) {
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if (toDelete == null) {
                    throw new MqException("删除失败, 因为不存在该交换机, exchangeName=" + exchangeName);
                }
                // 查看该交换机的绑定关系，这里规定不能删除存在绑定关系的交换机，必须先解绑，再删除
                ConcurrentHashMap<String, Binding> bindings = memoryDataCenter.getExchangeBindings(exchangeName);
                if (bindings == null || bindings.isEmpty()) {
                    if (toDelete.isDurable()) diskDataCenter.deleteExchange(exchangeName);
                    memoryDataCenter.deleteExchange(exchangeName);
                    log.info("交换机删除成功, exchangeName=" + exchangeName);
                    return true;
                } else {
                    throw new MqException("删除失败, 该交换机尚存在绑定关系, 请解除全部绑定后重试, exchangeName=" + exchangeName);
                }
            }
        } catch (Exception e) {
            log.info("交换机删除失败, exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    // 创建队列
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive,
                                boolean autoDelete, Map<String, Object> arguments) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                MSGQueue toDeclare = memoryDataCenter.getQueue(queueName);
                if (toDeclare != null) {
                    log.info("队列已经存在, queueName=" + queueName);
                    return true;
                }
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                if (durable) diskDataCenter.insertQueue(queue);
                memoryDataCenter.insertQueue(queue);
                log.info("队列创建成功, queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            log.info("队列创建失败, queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除队列
    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                MSGQueue toDelete = memoryDataCenter.getQueue(queueName);
                if (toDelete == null) {
                    throw new MqException("删除失败, 因为不存在该队列, queueName=" + queueName);
                }
                // 查看该队列的绑定关系，这里规定不能删除存在绑定关系的队列，必须先解绑，再删除
                ConcurrentHashMap<String, Binding> queueBindings = memoryDataCenter.getQueueBindings(queueName);
                if (queueBindings == null || queueBindings.isEmpty()) {
                    if (toDelete.isDurable()) diskDataCenter.deleteQueue(queueName);
                    memoryDataCenter.deleteQueue(queueName);
                    log.info("队列删除成功, queueName=" + queueName);
                    return true;
                } else {
                    throw new MqException("删除失败, 该队列尚存在绑定关系, 请解除全部绑定后重试, queueName=" + queueName);
                }
            }
        } catch (Exception e) {
            log.info("队列删除失败, queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 绑定
    public boolean queueBind(String exchangeName, String queueName, String bindingKey) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    Binding toBind = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (toBind != null) {
                        log.info("该绑定已存在, exchangeName=" + exchangeName + ", queueName=" + queueName);
                        return true;
                    }
                    // 验证 bindingKey 的合法性
                    if (!router.checkBindingKey(bindingKey))
                        throw new MqException("bindingKey 非法, bindingKey=" + bindingKey);
                    // 验证将要绑定的 exchange 和 queue 是否都存在
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MqException("不存在该交换机, exchangeName=" + exchangeName);
                    }
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("不存在该队列, queueName=" + queueName);
                    }
                    // 通过上述校验后，再进行对象的创建和插入
                    Binding binding = new Binding();
                    binding.setQueueName(queueName);
                    binding.setExchangeName(exchangeName);
                    binding.setBindingKey(bindingKey);
                    if (exchange.isDurable() && queue.isDurable()) diskDataCenter.insertBinding(binding);
                    memoryDataCenter.insertBinding(binding);
                    log.info("绑定成功, exchangeName=" + exchangeName + ", queueName=" + queueName);
                }
            }
            return true;
        } catch (Exception e) {
            log.info("绑定失败, exchangeName=" + exchangeName + ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 解绑
    public boolean queueUnbind(String exchangeName, String queueName) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    Binding toUnbind = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (toUnbind == null) throw new MqException("解除绑定失败, 因为不存在该绑定");
                    // 验证将要解除绑定的 exchange 和 queue 是否都存在
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MqException("不存在该交换机, exchangeName=" + exchangeName);
                    }
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("不存在该队列, queueName=" + queueName);
                    }
                    // 通过上述校验后，再进行绑定的解除
                    if (exchange.isDurable() && queue.isDurable()) diskDataCenter.deleteBinding(toUnbind);
                    memoryDataCenter.deleteBinding(toUnbind);
                    log.info("解除绑定成功, exchangeName=" + exchangeName + ", queueName=" + queueName);
                    return true;
                }
            }
        } catch (Exception e) {
            log.info("解除绑定失败, exchangeName=" + exchangeName + ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 发布消息（生产者先将消息交给交换机，由交换机根据绑定关系将消息放入相应队列）
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) {
        try {
            exchangeName = virtualHostName + exchangeName;
            // 校验 routingKey 的合法性
            if (!router.checkRoutingKey(routingKey)) throw new MqException("routingKey 非法, routingKey=" + routingKey);
            // 检查交换机是否存在
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) throw new MqException("交换机不存在, exchangeName=" + exchangeName);
            // 检查交换机类型
            if (exchange.getType() == ExchangeType.DIRECT) {
                // 对于直接交换机，routingKey 即为队列名，此时无需考虑绑定关系
                String queueName = virtualHostName + routingKey;
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                // 检查队列是否存在
                if (queue == null) throw new MqException("队列不存在, queueName=" + queueName);
                // 构造并发送消息
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                sendMessage(queue, message);
            } else {
                // 对于 FANOUT 和 TOPIC 交换机，都需要考虑绑定关系
                ConcurrentHashMap<String, Binding> bindings = memoryDataCenter.getExchangeBindings(exchangeName);
                if (bindings == null) throw new MqException("该交换机未绑定任何队列, exchangeName=" + exchangeName);
                // 利用 entry 取出所有绑定关系
                for (Map.Entry<String, Binding> entry : bindings.entrySet()) {
                    Binding binding = entry.getValue();
                    // 检查队列是否存在
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        log.warn("发送消息时, 部分队列不存在, queueName=" + binding.getQueueName());
                        continue;
                    }
                    // 构造消息
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    // FANOUT：队列已绑定 -> 转发
                    // TOPIC：队列已绑定 -> routingKey 和 bindingKey 能够匹配 -> 转发
                    // 若不能转发，直接跳过
                    if (!router.route(exchange.getType(), binding, message)) continue;
                    // 转发
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            log.info("消息发送失败");
            e.printStackTrace();
            return false;
        }
    }

    // 发送消息（将消息写入硬盘和内存）
    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        Integer deliverMode = message.getDeliverMode();
        // 写入硬盘：deliverMode 为 1 表示不持久化，为 2 表示持久化
        if (deliverMode == 2) diskDataCenter.sendMessage(queue, message);
        // 写入内存
        memoryDataCenter.sendMessage(queue, message);
        // 通知消费者可以消费消息了
        consumeManager.notifyConsume(queue.getName());
    }

    // 订阅消息（添加一个订阅者）
    public boolean basicConsume(String consumeTag, String queueName, boolean autoAck, Consumer consumer) {
        queueName = virtualHostName + queueName;
        try {
            consumeManager.addConsumer(consumeTag, queueName, autoAck, consumer);
            log.info("订阅消息成功，订阅队列为：" + queueName);
            return true;
        } catch (Exception e) {
            log.info("订阅消息失败，订阅队列为：" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 手动确认应答
    public boolean basicAck(String queueName, String messageId) {
        queueName = virtualHostName + queueName;
        try {
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null) throw new MqException("确认应答失败, 因为消息不存在, Message=" + messageId);
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) throw new MqException("确认应答失败, 因为队列不存在, queueName=" + queueName);
            // 删硬盘
            if (message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue, message);
            }
            // 删内存（待确认队列和消息中心）
            memoryDataCenter.removeMessageWaitAck(queue.getName(), message.getMessageId());
            memoryDataCenter.deleteMessage(message.getMessageId());
            log.info("已确认应答, messageId=" + messageId + ", queueName=" + queueName);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            log.info("应答失败, messageId=" + messageId + ", queueName=" + queueName);
            return false;
        }
    }
}
